/*
 * Copyright (c) 2017,安徽皖通科技股份有限公司 All rights reserved.
 */

package com.wttech.dzzf.process.service.transfer;

import com.wtkj.vo.dzzf.base.kfwd.FundAccount;
import com.wttech.dzzf.process.dao.DataTransferDao;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;

/**
 * 数据迁移分析入口服务实现
 *
 * @author MaShibo
 * @date 2017/11/6.
 */
@Service
public class DataTransferService {

    private static final Logger LOGGER = LoggerFactory.getLogger(DataTransferService.class);

    private static final Object LOCK = new Object();

    /**
     * 错误信息日志
     */
    static final Logger ERRORINFO = LoggerFactory.getLogger("ErrorInfo");

    /**
     * 默认的单次最大处理数量
     */
    private static final int DEFAULT_SINGLE_MAX_PROCESSNUM = 30000;

    /**
     * 默认的单次处理数量
     */
    private static final int DEFAULT_PROCESS_NUM = 10000;

    /**
     * 总数据量
     */
    private volatile int total;


    @Resource
    private DataTransferDao dataTransferDao;

    @Resource
    private FundAccountCheckService fundAccountCheckService;

    @Resource
    private ThreadPoolTaskExecutor executor;

    private void analyse() {
        total = dataTransferDao.queryFundAccountCount();
        LOGGER.info("共{}条待迁移的资金账户信息", total);
        if (total > DEFAULT_SINGLE_MAX_PROCESSNUM) {
            LOGGER.info("待迁移数据大于{}条，使用多线程处理", DEFAULT_SINGLE_MAX_PROCESSNUM);
            int count = (int) Math.ceil((double) total / DEFAULT_PROCESS_NUM);
            LOGGER.info("每个线程单次处理{}条数据，共需处理{}次", DEFAULT_PROCESS_NUM, count);
            List<Future<Void>> futures = new ArrayList<>(count);
            for (int i = 0; i < count; i++) {
                futures.add(executor.submit(new SubTransferThread(i + 1)));
            }
            futures.forEach(future -> {
                try {
                    future.get();
                } catch (Exception e) {
                    LOGGER.error(e.getMessage(), e);
                }
            });
        }
    }

    private class SubTransferThread implements Callable<Void> {

        private int count;

        SubTransferThread(int count) {
            this.count = count;
        }

        @Override
        public Void call() throws Exception {
            LOGGER.info("第{}次开始执行", count);
            int startRow = (count - 1) * DEFAULT_PROCESS_NUM + 1;
            int endRow = count * DEFAULT_PROCESS_NUM;
            List<FundAccount> fundAccounts = dataTransferDao.queryFundAccountInfs(startRow, endRow);
            fundAccounts.forEach(this::analyseSingleData);
            LOGGER.info("第{}次结束执行", count);
            return null;
        }

        private void analyseSingleData(FundAccount fundAccount) {
            try {
                fundAccountCheckService.analyseFundAccount(fundAccount);
            } catch (Exception e) {
                LOGGER.error(e.getMessage(), e);
            }
            synchronized (LOCK) {
                LOGGER.info("还有【{}】条数据待处理", --total);
            }
        }
    }

    public static void main(String[] args) {
        ApplicationContext context = new ClassPathXmlApplicationContext("applicationContext.xml");
        DataTransferService service = context.getBean(DataTransferService.class);
        service.analyse();
    }
}
